package com.atguigu.gmall.realtime.handler

import com.alibaba.fastjson.{JSON, JSONObject}
import com.atguigu.gmall.realtime.handler.DimHandler.CouponDim
import com.atguigu.gmall.realtime.util.{MyBeanUtil, RedisUtil}
import org.apache.spark.streaming.dstream.DStream
import redis.clients.jedis.Jedis

import java.time.{LocalDate, Period}
import scala.reflect.{ClassTag, classTag}

object DimHandler {



  //  def joinUser[T:ClassTag](dstream:DStream[T],keyByFunc: T => String): DStream[T] ={
  //
  //     joinUser[T](dstream,keyByFunc,
  //       {case (t:T,userDim: UserDim)=>
  //        MyBeanUtil.copyProperties(userDim,t.asInstanceOf[Object])
  //        t
  //     })
  //  }


  def joinUser[T: ClassTag](dstream: DStream[T],
                            keyByFunc: T => String,
                           ): DStream[T] = {
    dstream.map { case obj: T =>
      val userId: String = keyByFunc(obj)
      if (userId != null && userId != "0") {
        val jedisClient: Jedis = RedisUtil.getJedisClient

        val userKey = s"DIM:user_info:$userId"
        val userJson: String = jedisClient.get(userKey)
       // println("joinUser:"+userJson)
        val userJsonObj: JSONObject = JSON.parseObject(userJson)
        var user_gender=""
        if(userJsonObj.getString("gender")!=null&&userJsonObj.getString("gender").size>0){
            user_gender  = userJsonObj.getString("gender") match {
            case "M" => "男"
            case "F" => "女"
          }
        }

        val birthday: String = userJsonObj.getString("birthday")
        val birthdayDate: LocalDate = LocalDate.parse(birthday)
        val now: LocalDate = LocalDate.now()
        val period = Period.between(birthdayDate, now);
        val years: Int = period.getYears

        val user_age = years
        val userDim: UserDim = UserDim(user_gender, user_age)
        jedisClient.close()
        MyBeanUtil.copyProperties(userDim, obj.asInstanceOf[Object])
      }
      obj
    }
  }

  //  def joinProvince[T:ClassTag](dstream:DStream[T],keyByFunc: T => String): DStream[T] ={
  //
  //    joinProvince[T](dstream,keyByFunc,{case (t:T,provinceDim: ProvinceDim)=>
  //      MyBeanUtil.copyProperties(provinceDim,t.asInstanceOf[Object])
  //      t
  //    })
  //
  //  }


  def joinProvince[T: ClassTag](dstream: DStream[T],
                                keyByFunc: T => String,
                               ): DStream[T] = {
    dstream.map { obj =>
      val provinceId: String = keyByFunc(obj)
      if (provinceId != null && provinceId != "0") {

        val jedisClient: Jedis = RedisUtil.getJedisClient

        val provinceKey = s"DIM:base_province:$provinceId"
        //地区信息
        val provinceJson: String = jedisClient.get(provinceKey)
        val provinceJsonObj: JSONObject = JSON.parseObject(provinceJson)

        val provinceDim: ProvinceDim = ProvinceDim(
          provinceJsonObj.getString("name"),
          provinceJsonObj.getString("iso_code"),
          provinceJsonObj.getString("iso_3166_2"),
          provinceJsonObj.getString("area_code")
        )
        jedisClient.close()
        MyBeanUtil.copyProperties(provinceDim, obj.asInstanceOf[Object])
      }
      obj


    }
  }


  //  def joinSku[T:ClassTag](dstream:DStream[T],keyByFunc: T => String): DStream[T] ={
  //
  //    joinSku[T](dstream,keyByFunc,{case (t:T,skuDim: SkuDim)=>
  //
  //      t
  //    })
  //
  //  }

  def joinSku[T: ClassTag](dstream: DStream[T],
                           keyByFunc: T => String
                          ): DStream[T] = {
    dstream.map { obj =>
      val skuId: String = keyByFunc(obj)
      if (skuId != null && skuId != "0") {
        val jedisClient: Jedis = RedisUtil.getJedisClient

        val skuKey = s"DIM:sku_info:$skuId"
        val skuJson: String = jedisClient.get(skuKey)
        val skuJsonObj: JSONObject = JSON.parseObject(skuJson)


        val tmId: String = skuJsonObj.getString("tm_id")
        val tmKey = s"DIM:base_trademark:$tmId"
        val tmJson: String = jedisClient.get(tmKey)
        val tmJsonObj: JSONObject = JSON.parseObject(tmJson)

        val spuId: String = skuJsonObj.getString("spu_id")
        val spuKey = s"DIM:spu_info:$spuId"
        val spuJson: String = jedisClient.get(spuKey)
        val spuJsonObj: JSONObject = JSON.parseObject(spuJson)

        val c3Id: String = skuJsonObj.getString("category3_id")
        val c3Key = s"DIM:base_category3:$c3Id"
        val c3Json: String = jedisClient.get(c3Key)
        val c3JsonObj: JSONObject = JSON.parseObject(c3Json)


        val tm_name: String = tmJsonObj.getString("tm_name")
        val spu_name: String = spuJsonObj.getString("spu_name")
        val c3_name: String = c3JsonObj.getString("name")
        val sku_name: String = skuJsonObj.getString("sku_name")

        val skuDim: SkuDim = SkuDim(skuId.toLong, sku_name, tmId.toLong, tm_name, c3Id.toLong, c3_name, spuId.toLong, spu_name)

        jedisClient.close()
        MyBeanUtil.copyProperties(skuDim, obj.asInstanceOf[Object])
      }
      obj
    }
  }


  def joinCoupon[T: ClassTag](dstream: DStream[T],
                              keyByFunc: T => String
                             ): DStream[T] = {
    dstream.map { obj =>

      val couponId: String = keyByFunc(obj)
      if (couponId != null && couponId != "0") {
        val jedisClient: Jedis = RedisUtil.getJedisClient

        val couponKey = s"DIM:coupon_info:$couponId"
        val couponJson: String = jedisClient.get(couponKey)
        if(couponJson!=null&& couponJson.size>0){
          val couponJsonObj: JSONObject = JSON.parseObject(couponJson)

          val coupon_id: String = couponJsonObj.getString("id")
          val coupon_name: String = couponJsonObj.getString("coupon_name")
          val coupon_type: String = couponJsonObj.getString("coupon_type")
          val coupon_condition_amount: Double = couponJsonObj.getDouble("condition_amount")
          val coupon_condition_num: Long = couponJsonObj.getLong("condition_num")
          val couponDim: CouponDim = CouponDim(coupon_id.toLong, coupon_name, coupon_type, coupon_condition_amount, coupon_condition_num)
          MyBeanUtil.copyProperties(couponDim, obj.asInstanceOf[Object])
        }
        jedisClient.close()

      }
      obj
    }
  }



  //  def joinActivity[T:ClassTag](dstream:DStream[T],keyByFunc: T => String): DStream[T] ={
  //
  //    joinActivity[T](dstream,keyByFunc,{case (t:T,activityDim: ActivityDim)=>
  //      MyBeanUtil.copyProperties(activityDim,t.asInstanceOf[Object])
  //      t
  //    })
  //
  //  }

  def joinActivity[T: ClassTag](dstream: DStream[T],
                                keyByFunc: T => String
                               ): DStream[T] = {
    dstream.map { obj =>
      val activityRuleId: String = keyByFunc(obj)
      if (activityRuleId != null && activityRuleId != "0") {


        val jedisClient: Jedis = RedisUtil.getJedisClient

        val actRuleKey = s"DIM:activity_rule:$activityRuleId"

        val actRuleJson: String = jedisClient.get(actRuleKey)
        if(actRuleJson!=null && actRuleJson.size>0) {
          val actRuleJsonObj: JSONObject = JSON.parseObject(actRuleJson)


          val activity_rule_id: String = actRuleJsonObj.getString("id")
          val activity_id: String = actRuleJsonObj.getString("activity_id")
          val activity_type: String = actRuleJsonObj.getString("activity_type")
          val activity_condition_amount: Double = actRuleJsonObj.getDouble("condition_amount")
          val activity_condition_num: Long = actRuleJsonObj.getLong("condition_num")


          val actKey = s"DIM:activity_info:$activity_id"

          val actJson: String = jedisClient.get(actKey)
          val actJsonObj: JSONObject = JSON.parseObject(actJson)
          val activity_name: String = actJsonObj.getString("activity_name")

          val activityDim: ActivityDim = ActivityDim(activity_id.toLong, activity_name, activity_type, activity_condition_amount, activity_condition_num)
          MyBeanUtil.copyProperties(activityDim, obj.asInstanceOf[Object])
        }
        jedisClient.close()


      }
      obj
    }
  }


  def joinBaseDic[T: ClassTag](dstream: DStream[T],
                                keyByFunc: T => String,
                               matchFunc: (T,String)=> Unit,
                               ): DStream[T] = {
    dstream.map { obj =>
      val baseDicId: String = keyByFunc(obj)
      if (baseDicId != null && baseDicId != "0") {


        val jedisClient: Jedis = RedisUtil.getJedisClient

        val baseDicKey = s"DIM:base_dic:$baseDicId"

        val baseDicJson: String = jedisClient.get(baseDicKey)
        if(baseDicJson!=null && baseDicJson.size>0) {
          val baseDicDim: BaseDicDim = JSON.parseObject(baseDicJson, classOf[BaseDicDim])
          matchFunc(obj,baseDicDim.dic_name)
        }
        jedisClient.close()
      }
      obj
    }
  }


  case class CouponDim(var coupon_id: Long = 0L,
                       var coupon_name: String = null,
                       var coupon_type: String = null,
                       var coupon_condition_amount: Double = 0D,
                       var coupon_condition_num: Long = 0L) {

  }


  case class ActivityDim(var activity_id: Long = 0L,
                         var activity_name: String = null,
                         var activity_type: String = null,
                         var activity_condition_amount: Double = 0D,
                         var activity_condition_num: Long = 0L) {

  }

  case class UserDim(var user_gender: String, user_age: Int) {

  }

  case class BaseDicDim(  dic_name: String) {

  }

  case class SkuDim(var sku_id: Long,
                    var sku_name: String,
                    var tm_id: Long,
                    var tm_name: String,
                    var category3_id: Long,
                    var category3_name: String,
                    var spu_id: Long,
                    var spu_name: String)


  case class ProvinceDim(
                         var province_name: String, //名称 quickbi
                         var province_iso_code: String, // superset
                         var province_3166_2: String, //es
                         var province_area_code: String // datav
                        )
}
